// =============================================================
// CSpaceProxy
// ---------------------------------------
// Author : Woo-Hyun Kim
// Email : woorung@icu.ac.kr or woorung@nhncorp.com
// Date : Oct. 9, 2007
// =============================================================
#include "CSpaceProxy.h"

//---------------------------------------------------------------
// avoid cross reference
//---------------------------------------------------------------
#include "CWorkSpaces.h"
//---------------------------------------------------------------

//#define ONE_BY_ONE 1
//#define BIGGEST 1

CSpaceProxy::CSpaceProxy()
{
	_topology = NULL;
}

CSpaceProxy::CSpaceProxy(CWorkSpaces *wspace)
{
	_service = wspace->service();
	_instance = wspace->instance();
	_topology = wspace->topology();
}

CSpaceProxy::~CSpaceProxy()
{
}

int CSpaceProxy::workspace(CWorkSpaces *wspace)
{
	_service = wspace->service();
	_instance = wspace->instance();
	_topology = wspace->topology();
}

char *CSpaceProxy::getLocalIP()
{
	if(local_ip=="") {
		char name[HOST_NAME_MAX];
		int h_err;
		char buf[4096];
		struct hostent he, *h;
		struct sockaddr_in addr;

		if(gethostname(name,sizeof(name))==0) {
			// for thread-safety
			if(gethostbyname_r(name,&he,buf,4096,&h,&h_err)==0) {
				for(int i=0;he.h_addr_list[i]!=NULL;i++) {
					memcpy(&(addr.sin_addr), he.h_addr_list[i], he.h_length);
					local_ip = inet_ntoa(addr.sin_addr);

					// just return the first IP
					break;
				}
			}
		}
	}

	return (char *)local_ip.c_str();
}

long CSpaceProxy::size()
{
	string space_key = "";

	return size(space_key);
}

long CSpaceProxy::size(string &space_key)
{
	long total_size = 0;

	if(space_key=="") { // in order to get data size from "locally" distributed spaces
		// get local spaces
		vector<CServiceInfo> lspaces;
		lspaces = _topology->llookup(getLocalIP(), lspaces);

		vector< CSpaceMonitorThread * > mons;

		for(int i=0;i<lspaces.size();i++) {
			CSpaceMonitorThread *mon = new CSpaceMonitorThread(lspaces[i].getName());

			mons.push_back(mon);

			mons[i]->workspace(_service.c_str(), _instance.c_str());
			mons[i]->add(lspaces[i].getIP(), lspaces[i].getPort());

			mons[i]->start();
		}

		for(int i=0;i<lspaces.size();i++) {
			mons[i]->join();

			total_size += mons[i]->size();

			delete mons[i];
		}
	}
	else {
		CServiceInfo found_space = _topology->lookup((char *)space_key.c_str());
		CSpaceMonitorThread *mon = new CSpaceMonitorThread(found_space.getName());
		
		mon->workspace(_service.c_str(), _instance.c_str());
		mon->add(found_space.getIP(), found_space.getPort());

		mon->start();

		mon->join();
		
		total_size = mon->size();

		delete mon;
	}

	return total_size;
}

int CSpaceProxy::biggest(CServiceInfo &biggest_space)
{
	vector<CServiceInfo> lspaces;
	
	lspaces = _topology->llookup(getLocalIP(), lspaces);

	int bigger_idx = 0;
	int bigger_size = 0;

	vector< CSpaceMonitorThread * > mons;

	for(int i=0;i<lspaces.size();i++) {
		CSpaceMonitorThread *mon = new CSpaceMonitorThread(lspaces[i].getName());

		mons.push_back(mon);

		mons[i]->workspace(_service.c_str(), _instance.c_str());
		mons[i]->add(lspaces[i].getIP(), lspaces[i].getPort());

		mons[i]->start();
	}

	for(int i=0;i<lspaces.size();i++) {
		mons[i]->join();

		if(bigger_size<mons[i]->size()) {
			bigger_idx = i;
			bigger_size = mons[i]->size();
		}

		delete mons[i];
	}

	if(bigger_size>0) {
		biggest_space = lspaces[bigger_idx];
		return 1;
	}
	else return -1;
}

int CSpaceProxy::onebyone(CServiceInfo &onebyone_space)
{
	vector<CServiceInfo> lspaces;
	
	lspaces = _topology->llookup(getLocalIP(), lspaces);

	for(int i=0;i<lspaces.size();i++) {
		CSpaceMonitorThread mon(lspaces[i].getName());
		mon.workspace(_service.c_str(), _instance.c_str());
		mon.add(lspaces[i].getIP(), lspaces[i].getPort());

		mon.start();
		mon.join();

		if(mon.size()>0) {
			onebyone_space = lspaces[i];
			return 1;
		}
	}

	return -1;
}

int CSpaceProxy::randomly(CServiceInfo &randomly_space)
{
	vector<CServiceInfo> lspaces;

	lspaces = _topology->llookup(getLocalIP(), lspaces);

	int n = lspaces.size();

	struct timeb itb;
	struct tm *lt;
	ftime(&itb);
	lt = localtime(&itb.time);
	srand(lt->tm_sec+itb.millitm);

	int all_zero = n;
	int r = rand() % n;
	while(1) {
		CSpaceMonitorThread mon(lspaces[r].getName());
		mon.workspace(_service.c_str(), _instance.c_str());
		mon.add(lspaces[r].getIP(), lspaces[r].getPort());

		mon.start();
		mon.join();

		if(mon.size()>0) {
			randomly_space = lspaces[r];
			return 1;
		}
		else {
			if(--all_zero<1) break;

			r = (r+1) % n;
		}
	}

	return -1;
}

int CSpaceProxy::clean(string &key)
{
	// find a specific space responsible for the data key
	string space_key = key;
	// find a specific space responsible for the data key
	CServiceInfo found_space = _topology->lookup((char *) space_key.c_str());

#ifdef DEBUG
	cout << "####################################################" << endl;
	cout << "service name = " << found_space.getName() << endl;
	cout << "host ip = " << found_space.getIP() << endl;
	cout << "host port = " << found_space.getPort() << endl;
	cout << "####################################################" << endl;
#endif

	CServiceProxy space_proxy;
	space_proxy.connect(found_space.getIP(), found_space.getPort());

	cube__CService space_service;
	space_service.service(found_space.getName());
	space_service.instance((char *)_instance.c_str());

	int nothing;
	// remove the space service instance
	if(space_proxy.finalize(&space_service, nothing)!=SOAP_OK) {
		soap_print_fault(space_proxy.soap(), stderr);
	}
}

int CSpaceProxy::write(pair<string,string> &kv)
{
	string space_key = (char *)kv.first.c_str();

	return write(space_key, kv);
}

int CSpaceProxy::write(string &space_key, pair<string,string> &kv)
{
	char *key = (char *)kv.first.c_str();
	char *value = (char *)kv.second.c_str();

	char *cmd = CUBE_SPACE_WRITE;

	CObjectArray *pobjs = new CObjectArray(3);
	(*pobjs)[0] = new cube__CKeyValue();
	dynamic_cast<cube__CKeyValue*>((*pobjs)[0])->setKey(CUBE_SPACE_ACTION);
	dynamic_cast<cube__CKeyValue*>((*pobjs)[0])->setValue(cmd);
	// the end of parameters
	(*pobjs)[1] = NULL;

	(*pobjs)[2] = new cube__CKeyValue();
	dynamic_cast<cube__CKeyValue *>((*pobjs)[2])->set(key, value);

#ifdef DEBUG
	cout << "inserting key = " << key << endl;
	cout << "inserting value = " << value << endl;
#endif

	// find a specific space responsible for the data key
	CServiceInfo found_space = _topology->lookup((char *) space_key.c_str());

#ifdef DEBUG
	cout << "####################################################" << endl;
	cout << "service name = " << found_space.getName() << endl;
	cout << "host ip = " << found_space.getIP() << endl;
	cout << "host port = " << found_space.getPort() << endl;
	cout << "####################################################" << endl;
#endif

	CServiceProxy space_proxy;
	space_proxy.connect(found_space.getIP(), found_space.getPort());

	cube__CService space_service;
	space_service.service(found_space.getName());
	space_service.instance((char *)_instance.c_str());

	struct cube__invokeResponse res;

	if(space_proxy.invoke(&space_service, pobjs, res)!=SOAP_OK) {
		soap_print_fault(space_proxy.soap(), stderr);

		delete (*pobjs)[0];
		// delete (*pobjs)[1]; // 1 is NULL
		delete (*pobjs)[2];
		delete pobjs;
	}

	delete (*pobjs)[0];
	// delete (*pobjs)[1]; // 1 is NULL
	delete (*pobjs)[2];
	delete pobjs;
}

int CSpaceProxy::read(pair<string,string> &kv)
{
	string space_key = (char *)kv.first.c_str();

	return read(space_key, kv);
}

int CSpaceProxy::read(string &space_key, pair<string,string> &kv)
{
	char *cmd = CUBE_SPACE_READ;

	return read(space_key, kv, cmd);
}

int CSpaceProxy::readIfExists(pair<string,string> &kv)
{
	string space_key = (char *)kv.first.c_str();

	return readIfExists(space_key, kv);
}

int CSpaceProxy::readIfExists(string &space_key, pair<string,string> &kv)
{
	char *cmd = CUBE_SPACE_READ_IF_EXISTS;

	return read(space_key, kv, cmd);
}

int CSpaceProxy::read(string &space_key, pair<string,string> &kv, char *cmd)
{
	CObjectArray *pobjs = new CObjectArray(3);
	(*pobjs)[0] = new cube__CKeyValue();
	dynamic_cast<cube__CKeyValue *>((*pobjs)[0])->setKey(CUBE_SPACE_ACTION);
	dynamic_cast<cube__CKeyValue *>((*pobjs)[0])->setValue(cmd);
	// the end of parameters
	(*pobjs)[1] = NULL;

	CServiceInfo found_space;
	if(space_key=="") { // read any key from locally distributed spaces
		(*pobjs)[2] = NULL;

#ifdef ONE_BY_ONE
		// lookup the onebyone space in local spaces
		if(onebyone(found_space)<0) {
#ifdef DEBUG
			cerr << endl;
			cerr << "### [onebyone]what the heck~!!!!!!" << endl;
			cerr << endl;
#endif
			delete (*pobjs)[0];
			// delete (*pobjs)[1]; // 1 is NULL
			delete pobjs;

			return -1;
		}
#else

#ifdef BIGGEST

		// lookup the biggest space in local spaces
		if(biggest(found_space)<0) {
#ifdef DEBUG
			cerr << endl;
			cerr << "### [biggest]what the heck~!!!!!!" << endl;
			cerr << endl;
#endif
			delete (*pobjs)[0];
			// delete (*pobjs)[1]; // 1 is NULL
			delete pobjs;

			return -1;
		}
#else
		// lookup the randomly space in local spaces
		if(randomly(found_space)<0) {
#ifdef DEBUG
			cerr << endl;
			cerr << "### [randomly]what the heck~!!!!!!" << endl;
			cerr << endl;
#endif
			delete (*pobjs)[0];
			// delete (*pobjs)[1]; // 1 is NULL
			delete pobjs;

			return -1;
		}
#endif // biggest

#endif // one_by_one

	}
	else { // read a specific key from a corresponding space
		(*pobjs)[2] = new cube__CKeyValue();
		dynamic_cast<cube__CKeyValue *>((*pobjs)[2])->setKey((char *)kv.first.c_str());

		found_space = _topology->lookup((char *)space_key.c_str());
	}

#ifdef DEBUG
	cout << "####################################################" << endl;
	cout << "service name = " << found_space.getName() << endl;
	cout << "host ip = " << found_space.getIP() << endl;
	cout << "host port = " << found_space.getPort() << endl;
	cout << "####################################################" << endl;
#endif

	// create a proxy for the space
	CServiceProxy space_proxy;
	space_proxy.connect(found_space.getIP(), found_space.getPort());

	cube__CService space_service;
	space_service.service(found_space.getName());
	space_service.instance((char *)_instance.c_str());

	struct cube__invokeResponse res;

	if(space_proxy.invoke(&space_service, pobjs, res)!=SOAP_OK) {
		soap_print_fault(space_proxy.soap(), stderr);

		delete (*pobjs)[0];
		// delete (*pobjs)[1]; // 1 is NULL
		delete (*pobjs)[2];
		delete pobjs;

		return -1;
	}

	CObjectArray *array = dynamic_cast<CObjectArray *>(res.out);
	if(array!=NULL) {
		cube__CKeyValue *keyval = dynamic_cast<cube__CKeyValue *>(array->__ptr[0]);
		if(keyval==NULL) {
			cout << "not key-value type" << endl;
			delete (*pobjs)[0];
			// delete (*pobjs)[1]; // 1 is NULL
			delete (*pobjs)[2];
			delete pobjs;

			return -1;
		}

#ifdef DEBUG
		cout << "found key = " << keyval->getKey() << endl;
		cout << "found value = " << keyval->getValue() << endl;
#endif

		kv.first = keyval->getKey();
		kv.second = keyval->getValue();
	}
	else {
		delete (*pobjs)[0];
		// delete (*pobjs)[1]; // 1 is NULL
		delete (*pobjs)[2];
		delete pobjs;
		return -1;
	}

	delete (*pobjs)[0];
	// delete (*pobjs)[1]; // 1 is NULL
	delete (*pobjs)[2];
	delete pobjs;

	return 1;
}

int CSpaceProxy::take(pair<string,string> &kv)
{
	string space_key = (char *)kv.first.c_str();

	return take(space_key, kv);
}

int CSpaceProxy::take(string &space_key, pair<string,string> &kv)
{
	char *cmd = CUBE_SPACE_TAKE;

	return take(space_key, kv, cmd);
}

int CSpaceProxy::takeIfExists(pair<string,string> &kv)
{
	string space_key = (char *)kv.first.c_str();

	return takeIfExists(space_key, kv);
}

int CSpaceProxy::takeIfExists(string &space_key, pair<string,string> &kv)
{
	char *cmd = CUBE_SPACE_TAKE_IF_EXISTS;

	return take(space_key, kv, cmd);
}

int CSpaceProxy::take(string &space_key, pair<string,string> &kv, char *cmd)
{
	CObjectArray *pobjs = new CObjectArray(3);
	(*pobjs)[0] = new cube__CKeyValue();
	dynamic_cast<cube__CKeyValue *>((*pobjs)[0])->setKey(CUBE_SPACE_ACTION);
	dynamic_cast<cube__CKeyValue *>((*pobjs)[0])->setValue(cmd);
	// the end of parameters
	(*pobjs)[1] = NULL;

	CServiceInfo found_space;
	if(space_key=="") { // take any key from "locally" distributed spaces
		(*pobjs)[2] = NULL;

#ifdef ONE_BY_ONE
		// lookup the onebyone space in local spaces
		if(onebyone(found_space)<0) {
#ifdef DEBUG
			cerr << endl;
			cerr << "### [onebyone]what the heck~!!!!!!" << endl;
			cerr << endl;
#endif
			delete (*pobjs)[0];
			// delete (*pobjs)[1]; // 1 is NULL
			delete pobjs;

			return -1;
		}
#else

#ifdef BIGGEST
		// lookup the biggest space in local spaces
		if(biggest(found_space)<0) {
#ifdef DEBUG
			cerr << endl;
			cerr << "### [biggest]what the heck~!!!!!!" << endl;
			cerr << endl;
#endif
			delete (*pobjs)[0];
			// delete (*pobjs)[1]; // 1 is NULL
			delete pobjs;

			return -1;
		}
#else
		// lookup the randomly space in local spaces
		if(randomly(found_space)<0) {
#ifdef DEBUG
			cerr << endl;
			cerr << "### [randomly]what the heck~!!!!!!" << endl;
			cerr << endl;
#endif
			delete (*pobjs)[0];
			// delete (*pobjs)[1]; // 1 is NULL
			delete pobjs;

			return -1;
		}

#endif // biggest

#endif // one_by_one

	}
	else { // take a specific key from a corresponding space
		(*pobjs)[2] = new cube__CKeyValue();
		dynamic_cast<cube__CKeyValue *>((*pobjs)[2])->setKey((char *)kv.first.c_str());

		found_space = _topology->lookup((char *)space_key.c_str());
	}

#ifdef DEBUG
	cout << "####################################################" << endl;
	cout << "service name = " << found_space.getName() << endl;
	cout << "host ip = " << found_space.getIP() << endl;
	cout << "host port = " << found_space.getPort() << endl;
	cout << "####################################################" << endl;
#endif

	// create a proxy for the space
	CServiceProxy space_proxy;
	space_proxy.connect(found_space.getIP(), found_space.getPort());

	cube__CService space_service;
	space_service.service(found_space.getName());
	space_service.instance((char *)_instance.c_str());

	struct cube__invokeResponse res;

	if(space_proxy.invoke(&space_service, pobjs, res)!=SOAP_OK) {
		soap_print_fault(space_proxy.soap(), stderr);

		delete (*pobjs)[0];
		// delete (*pobjs)[1]; // 1 is NULL
		delete (*pobjs)[2];
		delete pobjs;

		return -1;
	}

	CObjectArray *array = dynamic_cast<CObjectArray *>(res.out);
	if(array!=NULL) {
		cube__CKeyValue *keyval = dynamic_cast<cube__CKeyValue *>(array->__ptr[0]);
		if(keyval==NULL) {
			cout << "not key-value type" << endl;
			delete (*pobjs)[0];
			// delete (*pobjs)[1]; // 1 is NULL
			delete (*pobjs)[2];
			delete pobjs;

			return -1;
		}

#ifdef DEBUG
		cout << "found key = " << keyval->getKey() << endl;
		cout << "found value = " << keyval->getValue() << endl;
#endif

		kv.first = keyval->getKey();
		kv.second = keyval->getValue();
	}
	else {
		delete (*pobjs)[0];
		// delete (*pobjs)[1]; // 1 is NULL
		delete (*pobjs)[2];
		delete pobjs;
		return -1;
	}

	delete (*pobjs)[0];
	// delete (*pobjs)[1]; // 1 is NULL
	delete (*pobjs)[2];
	delete pobjs;

	return 1;
}
